package com.seedog.mq.server;

import com.seedog.mq.core.MessageQueue;
import com.seedog.mq.core.MessageQueueImpl;
import com.seedog.mq.distributed.ClusterManager;
import com.seedog.mq.distributed.ClusterManagerImpl;
import com.seedog.mq.distributed.DistributedConfig;
import com.seedog.mq.distributed.MessageReplicator;
import com.seedog.mq.distributed.MessageReplicatorImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.Executors;

/**
 * 消息队列服务器主类，负责启动和管理SeeDog消息队列服务
 */
public class MessageQueueServer {
    private static final Logger logger = LoggerFactory.getLogger(MessageQueueServer.class);
    
    // 消息队列核心组件
    private MessageQueue messageQueue;
    
    // 集群管理组件
    private ClusterManager clusterManager;
    
    // 消息复制组件
    private MessageReplicator messageReplicator;
    
    // 服务器配置
    private ServerConfig config;
    
    // 执行器服务
    private ExecutorService executorService;
    
    // 服务器运行状态标志
    private volatile boolean running = false;
    
    // 网络服务器组件
    private NetworkServer networkServer;
    
    /**
     * 构造函数
     * @param config 服务器配置
     */
    public MessageQueueServer(ServerConfig config) {
        this.config = config != null ? config : ServerConfig.getDefault();
    }
    
    /**
     * 启动消息队列服务器
     */
    public void start() {
        if (running) {
            logger.warn("Message queue server is already running");
            return;
        }
        
        try {
            logger.info("Starting SeeDog Message Queue Server...");
            
            // 初始化执行器服务
            executorService = Executors.newFixedThreadPool(config.getThreadPoolSize());
            
            // 初始化消息队列核心组件
            messageQueue = new MessageQueueImpl();
            
            // 如果启用了分布式模式，初始化分布式组件
            if (config.isDistributedMode()) {
                initializeDistributedComponents();
            }
            
            // 初始化网络服务器
            networkServer = new NetworkServer(config.getNodePort());
            executorService.submit(networkServer);
            
            // 注册JVM关闭钩子
            registerShutdownHook();
            
            running = true;
            logger.info("SeeDog Message Queue Server started successfully");
            logger.info("Server configuration: {}", config);
            
        } catch (Exception e) {
            logger.error("Failed to start message queue server", e);
            shutdown();
            throw new RuntimeException("Failed to start message queue server", e);
        }
    }
    
    /**
     * 初始化分布式组件
     */
    private void initializeDistributedComponents() {
        try {
            logger.info("Initializing distributed components...");
            
            // 创建分布式配置
            DistributedConfig distributedConfig = new DistributedConfig.Builder()
                    .host(config.getNodeHost())
                    .port(config.getNodePort())
                    .replicationFactor(config.getReplicationFactor())
                    .build();
            
            // 初始化集群管理器
            clusterManager = new ClusterManagerImpl();
            clusterManager.initialize(distributedConfig);
            clusterManager.start();
            
            // 初始化消息复制器
            messageReplicator = new MessageReplicatorImpl();
            messageReplicator.initialize(distributedConfig, clusterManager);
            messageReplicator.start();
            
            logger.info("Distributed components initialized successfully");
        } catch (Exception e) {
            logger.error("Failed to initialize distributed components", e);
            throw new RuntimeException("Failed to initialize distributed components", e);
        }
    }
    
    /**
     * 注册JVM关闭钩子
     */
    private void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            logger.info("Received shutdown signal, preparing to stop server...");
            shutdown();
        }));
    }
    
    /**
     * 关闭消息队列服务器
     */
    public void shutdown() {
        if (!running) {
            return;
        }
        
        logger.info("Shutting down SeeDog Message Queue Server...");
        
        running = false;
        
        // 关闭网络服务器
        if (networkServer != null) {
            try {
                networkServer.shutdown();
            } catch (Exception e) {
                logger.error("Error stopping network server", e);
            }
        }
        
        // 关闭分布式组件
        if (messageReplicator != null) {
            try {
                messageReplicator.stop();
            } catch (Exception e) {
                logger.error("Error stopping message replicator", e);
            }
        }
        
        if (clusterManager != null) {
            try {
                clusterManager.stop();
            } catch (Exception e) {
                logger.error("Error stopping cluster manager", e);
            }
        }
        
        // 关闭执行器服务
        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                executorService.shutdownNow();
            }
        }
        
        logger.info("SeeDog Message Queue Server shutdown completed");
    }
    
    /**
     * 获取消息队列实例
     * @return 消息队列实例
     */
    public MessageQueue getMessageQueue() {
        return messageQueue;
    }
    
    /**
     * 获取集群管理器实例
     * @return 集群管理器实例
     */
    public ClusterManager getClusterManager() {
        return clusterManager;
    }
    
    /**
     * 获取消息复制器实例
     * @return 消息复制器实例
     */
    public MessageReplicator getMessageReplicator() {
        return messageReplicator;
    }
    
    /**
     * 检查服务器是否正在运行
     * @return 如果服务器正在运行则返回true，否则返回false
     */
    public boolean isRunning() {
        return running;
    }
    
    /**
     * 获取服务器配置
     * @return 服务器配置对象
     */
    public ServerConfig getConfig() {
        return config;
    }
    
    /**
     * 主方法，用于启动消息队列服务器
     */
    public static void main(String[] args) {
        try {
            // 创建默认服务器配置
            ServerConfig config = ServerConfig.getDefault();
            
            // 解析命令行参数
            parseCommandLineArgs(args, config);
            
            // 创建并启动消息队列服务器
            MessageQueueServer server = new MessageQueueServer(config);
            server.start();
            
            // 保持主线程运行，直到收到关闭信号
            CountDownLatch latch = new CountDownLatch(1);
            latch.await();
            
        } catch (Exception e) {
            logger.error("Fatal error in message queue server", e);
            System.exit(1);
        }
    }
    
    /**
     * 解析命令行参数
     */
    private static void parseCommandLineArgs(String[] args, ServerConfig config) {
        if (args == null || args.length == 0) {
            return;
        }
        
        for (int i = 0; i < args.length; i++) {
            String arg = args[i];
            try {
                switch (arg) {
                    case "--port":
                    case "-p":
                        if (i + 1 < args.length) {
                            config.setNodePort(Integer.parseInt(args[i + 1]));
                            i++;
                        }
                        break;
                    case "--host":
                    case "-h":
                        if (i + 1 < args.length) {
                            config.setNodeHost(args[i + 1]);
                            i++;
                        }
                        break;
                    case "--distributed":
                    case "-d":
                        config.setDistributedMode(true);
                        break;
                    case "--replication-factor":
                    case "-rf":
                        if (i + 1 < args.length) {
                            config.setReplicationFactor(Integer.parseInt(args[i + 1]));
                            i++;
                        }
                        break;
                    default:
                        logger.warn("Unknown command line argument: {}", arg);
                        break;
                }
            } catch (Exception e) {
                logger.error("Error parsing command line argument: {}", arg, e);
            }
        }
    }
    
    /**
     * 网络服务器内部类，用于处理客户端连接和请求
     */
    private class NetworkServer implements Runnable {
        private final int port;
        private volatile boolean running = false;
        private ServerSocket serverSocket;
        private ExecutorService clientExecutor;
        
        public NetworkServer(int port) {
            this.port = port;
            this.clientExecutor = Executors.newCachedThreadPool();
        }
        
        @Override
        public void run() {
            try {
                serverSocket = new ServerSocket(port);
                running = true;
                logger.info("Network server started on port {}", port);
                
                while (running && !Thread.currentThread().isInterrupted()) {
                    try {
                        // 接受客户端连接
                        Socket clientSocket = serverSocket.accept();
                        logger.debug("Accepted connection from {}", clientSocket.getRemoteSocketAddress());
                        
                        // 为每个客户端创建一个线程处理请求
                        clientExecutor.submit(() -> handleClient(clientSocket));
                    } catch (IOException e) {
                        if (running) {
                            logger.error("Error accepting client connection", e);
                        }
                    }
                }
            } catch (IOException e) {
                logger.error("Failed to start network server on port {}", port, e);
            } finally {
                shutdown();
            }
        }
        
        private void handleClient(Socket clientSocket) {
            try (BufferedReader in = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
                 PrintWriter out = new PrintWriter(clientSocket.getOutputStream(), true)) {
                
                // 读取客户端请求
                String request = in.readLine();
                if (request == null) {
                    return;
                }
                
                logger.debug("Received request: {}", request);
                
                // 处理请求并发送响应
                String response = processRequest(request);
                out.println(response);
                
            } catch (IOException e) {
                logger.error("Error handling client request", e);
            } finally {
                try {
                    clientSocket.close();
                } catch (IOException e) {
                    logger.error("Error closing client socket", e);
                }
            }
        }
        
        private String processRequest(String request) {
            if (request == null || request.trim().isEmpty()) {
                return "EMPTY_REQUEST";
            }
            
            // 解析请求命令
            String[] parts = request.split(" ", 2);
            String command = parts[0].toUpperCase();
            
            if ("STATUS".equals(command)) {
                // 返回服务器状态
                return "RUNNING: " + isRunning() + ", TOPICS: " + (messageQueue != null ? messageQueue.getAllTopics().size() : 0);
            } else if ("TOPICS".equals(command)) {
                // 返回所有主题
                if (messageQueue != null) {
                    return String.join(",", messageQueue.getAllTopics());
                } else {
                    return "No topics available";
                }
            } else if ("CREATE_TOPIC".equals(command) && parts.length > 1) {
                // 创建主题
                String topicName = parts[1];
                if (messageQueue != null) {
                    boolean created = messageQueue.createTopic(topicName);
                    return created ? "TOPIC_CREATED" : "TOPIC_EXISTS";
                } else {
                    return "SERVER_ERROR";
                }
            } else if ("DELETE_TOPIC".equals(command) && parts.length > 1) {
                // 删除主题
                String topicName = parts[1];
                if (messageQueue != null) {
                    boolean deleted = messageQueue.deleteTopic(topicName);
                    return deleted ? "TOPIC_DELETED" : "TOPIC_NOT_FOUND";
                } else {
                    return "SERVER_ERROR";
                }
            }
            return "UNKNOWN_COMMAND";
        }
        
        public void shutdown() {
            running = false;
            
            if (serverSocket != null && !serverSocket.isClosed()) {
                try {
                    serverSocket.close();
                } catch (IOException e) {
                    logger.error("Error closing server socket", e);
                }
            }
            
            if (clientExecutor != null && !clientExecutor.isShutdown()) {
                clientExecutor.shutdown();
                try {
                    if (!clientExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
                        clientExecutor.shutdownNow();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    clientExecutor.shutdownNow();
                }
            }
            
            logger.info("Network server shutdown completed");
        }
    }
}